use std::net::{TcpListener, TcpStream};
use std::thread;
use std::time::Duration;

use anyhow::{bail, ensure, Result};
use async_channel::{unbounded, Receiver, Sender};
use async_dup::Arc;
use futures::future::Either;
use futures::prelude::*;
use futures::{AsyncReadExt, AsyncWriteExt};
use smol::{Async, Task, Timer};
use toml::Value;

pub(crate) async fn listen(cfg: Value) -> Result<()> {
    let own_addr = cfg["own"]["addr"]
        .as_str()
        .expect("read config fail: own.addr");
    println!("{:?} > listen to {}", thread::current().id(), own_addr);
    let listener = Async::<TcpListener>::bind(own_addr)?;
    let mut incoming = listener.incoming();

    if cfg["gateway"]["mock"]
        .as_bool()
        .expect("read config fail: gateway.mock")
    {
        let config = cfg.clone();
        Task::spawn(async {
            if let Err(e) = mock_gateway(config).await {
                println!("mock gateway server: {}", e)
            }
        })
        .detach();
    }

    let (s, r) = unbounded::<String>();
    let config = cfg.clone();
    let sender = s.clone();
    //调用checker
    Task::spawn(async {
        prepare_check(config, sender, r)
            .await
            .expect("Call checker error!");
    })
    .detach();

    while let Some(stream) = incoming.next().await {
        let outside = Arc::new(stream?);
        let config = cfg.clone();
        let sender = s.clone();
        // let checker_s
        Task::spawn(async {
            if let Err(e) = shunt(config, outside, sender).await {
                println!("shunt: {}", e)
            }
        })
        .detach();
    }
    Ok(())
}

async fn shunt(
    cfg: Value,
    mut outside: Arc<Async<TcpStream>>,
    sender: Sender<String>,
) -> Result<()> {
    let time = std::time::SystemTime::now();
    let dur = Duration::from_secs(cfg["timeout_dur"].as_str().unwrap_or("30").parse()?);

    //接收
    let (len, msg) = read_timeout(dur, outside.clone()).await?;
    println!(
        "{:?} > receive over, time passed {}ms > [{}{}]",
        thread::current().id(),
        time.elapsed()?.as_millis(),
        String::from_utf8_lossy(&len),
        String::from_utf8_lossy(&msg)
    );

    //分流
    sender.send(String::from_utf8_lossy(&msg).into()).await?;

    //调用网关
    let (len, msg) = call_gateway(cfg, len, msg).await?;

    //响应
    outside.write_all(&len).await?;
    outside.write_all(&msg).await?;
    outside.flush().await?;
    outside.close().await?;

    println!(
        "{:?} > handle over, time passed {}ms > [{}{}]",
        thread::current().id(),
        time.elapsed()?.as_millis(),
        String::from_utf8_lossy(&len),
        String::from_utf8_lossy(&msg)
    );

    Ok(())
}

async fn conn_checker(cfg: &Value) -> Result<Async<TcpStream>> {
    //连接Checker
    let checker_addr = cfg["checker"]["addr"]
        .as_str()
        .expect("read config fail: checker.addr");
    Ok(Async::<TcpStream>::connect(checker_addr).await?)
}

async fn prepare_check(
    cfg: Value,
    sender: Sender<String>,
    receiver: Receiver<String>,
) -> Result<()> {
    let mut stream = conn_checker(&cfg).await?;
    loop {
        let msg = receiver.recv().await?;
        println!("{:?} > call checker > [{}]", thread::current().id(), msg);
        if let Err(e) = call_checker(&mut stream, &*msg).await {
            println!("call checker fail: {:?}", e);
            stream = conn_checker(&cfg).await?;
            let sender = sender.clone();
            Task::blocking(async move {
                Timer::after(Duration::from_secs(1)).await;
                sender.send(msg).now_or_never();
            })
            .detach();
        }
    }
}

async fn call_checker(
    stream: &mut Async<TcpStream>,
    msg: &str,
) -> std::result::Result<(), std::io::Error> {
    stream
        .write_all(format!("{:010}", msg.len()).as_bytes())
        .await?;
    stream.write_all(msg.as_bytes()).await?;
    stream.flush().await
}

async fn call_gateway(cfg: Value, len: [u8; 10], msg: Vec<u8>) -> Result<([u8; 10], Vec<u8>)> {
    //连接网关
    let gateway_addr = cfg["gateway"]["addr"]
        .as_str()
        .expect("read config fail: gateway.addr");
    let mut inside = Arc::new(Async::<TcpStream>::connect(gateway_addr).await?);

    //写网关
    inside.write_all(&len).await?;
    inside.write_all(&msg).await?;
    inside.flush().await?;

    //读网关
    let dur = Duration::from_secs(cfg["timeout_dur"].as_str().unwrap_or("30").parse()?);
    let res = read_timeout(dur, inside.clone()).await?;
    inside.close().await?;
    Ok(res)
}

async fn read_timeout(
    dur: Duration,
    mut reader: Arc<Async<TcpStream>>,
) -> Result<([u8; 10], Vec<u8>)> {
    // let time = std::time::SystemTime::now();
    let mut len = [0_u8; 10];
    timeout(dur, reader.read_exact(&mut len)).await??;
    let length: usize = String::from_utf8_lossy(&len).parse()?;
    if length < 1 {
        bail!("length is 0")
    }
    // println!(
    //     "{:?} > read length[{}] time passed {}ms",
    //     thread::current().id(),
    //     length,
    //     time.elapsed()?.as_millis()
    // );

    let mut msg = vec![0_u8; length];
    timeout(dur, reader.read_exact(&mut msg)).await??;
    // println!(
    //     "{:?} > [{}]th read, time passed {}ms",
    //     thread::current().id(),
    //     i + 1,
    //     time.elapsed()?.as_millis()
    // );

    ensure!(
        msg.len() == length,
        "Message length abnormal, real[{}] != head[{}]",
        msg.len(),
        length
    );

    Ok((len, msg))
}

async fn mock_gateway(cfg: Value) -> Result<()> {
    let gateway_addr = cfg["gateway"]["addr"]
        .as_str()
        .expect("read config fail: gateway.addr");
    let dur = Duration::from_secs(cfg["timeout_dur"].as_str().unwrap_or("30").parse()?);
    println!(
        "{:?} > mock gateway listen {}",
        thread::current().id(),
        gateway_addr
    );

    let listener = Async::<TcpListener>::bind(gateway_addr)?;
    let mut incoming = listener.incoming();
    while let Some(stream) = incoming.next().await {
        let stream: Async<TcpStream> = stream?;
        Task::spawn(async move {
            // println!("{:?} > gateway handle", thread::current().id());
            let time = std::time::SystemTime::now();

            let mut rw = Arc::new(stream);

            //read req
            let (len, msg) = read_timeout(dur, rw.clone())
                .await
                .expect("mock gateway read_msg fail");

            //mock business
            // println!("{:?} > gateway handle wait", thread::current().id());
            // Timer::after(Duration::from_secs(2)).await;

            //write res
            rw.write_all(&len)
                .await
                .expect("mock gateway write len fail");
            rw.write_all(&msg)
                .await
                .expect("mock gateway write msg fail");
            rw.flush().await.expect("mock gateway flush fail");
            rw.close().await.expect("mock gateway close tcp fail");
            println!(
                "{:?} > mock gateway handle over, time passed {}ms",
                thread::current().id(),
                time.elapsed().unwrap_or_default().as_millis()
            );
        })
        .detach();
    }
    Ok(())
}

async fn timeout<T>(dur: Duration, f: impl Future<Output = T>) -> Result<T> {
    futures::pin_mut!(f);
    match future::select(f, Timer::after(dur)).await {
        Either::Left((out, _)) => Ok(out),
        Either::Right(_) => anyhow::bail!("Timeout!!!"),
    }
}
